{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8bc47910-45c3-433e-a865-a6ef4a7af366",
   "metadata": {},
   "source": [
    "## A02_Data Formatting Process\n",
    "\n",
    "This code will convert the raw data into a formatted DataFrame and store it into a formatted Parquet file."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf3f9e86-9aaf-4886-8b53-19895cee8529",
   "metadata": {},
   "source": [
    "### 1. init spark session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "f4d85c4c-9c2c-4975-8c51-4837169df20e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "627b1415-b963-4ecf-b6f9-be9844eb56c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName(\"price_opendata\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9e2954b3-47c7-40f4-bc98-4e61c1e83229",
   "metadata": {},
   "source": [
    "### 2. read the json data files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "d08a99aa-a77c-4bd5-81d8-1c4acdefc793",
   "metadata": {},
   "outputs": [],
   "source": [
    "data_dir = \"./Landing Zone/price_opendata\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "d16b4b9a-8386-41e2-9a91-20449eb1e52c",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.json(data_dir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "eabf3844-4a3f-495e-af14-475ed518b89b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+--------------+--------------------+--------------------+\n",
      "|_id|district_id| district_name|                info|         neigh_name |\n",
      "+---+-----------+--------------+--------------------+--------------------+\n",
      "|  1|          1|  Ciutat Vella|[{97.0, 1726.5, N...|            el Raval|\n",
      "|  2|          1|  Ciutat Vella|[{212.4, 2448.2, ...|      el Barri Gòtic|\n",
      "|  3|          1|  Ciutat Vella|[{175.5, 3476.3, ...|      la Barceloneta|\n",
      "|  4|          1|  Ciutat Vella|[{190.4, 2846.5, ...|Sant Pere, Santa ...|\n",
      "|  5|          2|      Eixample|[{268.3, 2842.9, ...|       el Fort Pienc|\n",
      "|  6|          2|      Eixample|[{192.4, 2938.7, ...|  la Sagrada Família|\n",
      "|  7|          2|      Eixample|[{306.9, 2798.2, ...|la Dreta de l'Eix...|\n",
      "|  8|          2|      Eixample|[{239.2, 2760.2, ...|l'Antiga Esquerra...|\n",
      "|  9|          2|      Eixample|[{301.6, 3648.2, ...|la Nova Esquerra ...|\n",
      "| 10|          2|      Eixample|[{177.5, 2024.9, ...|         Sant Antoni|\n",
      "| 11|          3|Sants-Montjuïc|[{93.7, 1582.1, N...|        el Poble Sec|\n",
      "| 12|          3|Sants-Montjuïc|[{166.2, 2405.2, ...|   la Marina de Port|\n",
      "| 13|          3|Sants-Montjuïc|[{179.3, 2543.5, ...|la Font de la Gua...|\n",
      "| 14|          3|Sants-Montjuïc|[{177.8, 1834.2, ...|         Hostafrancs|\n",
      "| 15|          3|Sants-Montjuïc|[{102.4, 1960.8, ...|          la Bordeta|\n",
      "| 16|          3|Sants-Montjuïc|[{190.8, 3186.4, ...|       Sants - Badal|\n",
      "| 17|          3|Sants-Montjuïc|[{154.5, 2270.1, ...|               Sants|\n",
      "| 18|          4|     Les Corts|[{252.5, 2923.6, ...|           les Corts|\n",
      "| 19|          4|     Les Corts|[{266.7, 2890.5, ...|la Maternitat i S...|\n",
      "| 20|          4|     Les Corts|[{733.9, 4406.9, ...|           Pedralbes|\n",
      "+---+-----------+--------------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "437ba981-5e9f-4acb-951f-a4c9bee275bd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _id: long (nullable = true)\n",
      " |-- district_id: long (nullable = true)\n",
      " |-- district_name: string (nullable = true)\n",
      " |-- info: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- Amount: double (nullable = true)\n",
      " |    |    |-- PerMeter: double (nullable = true)\n",
      " |    |    |-- diffAmount: double (nullable = true)\n",
      " |    |    |-- diffPerMeter: double (nullable = true)\n",
      " |    |    |-- usedAmount: double (nullable = true)\n",
      " |    |    |-- usedPerMeter: double (nullable = true)\n",
      " |    |    |-- year: long (nullable = true)\n",
      " |-- neigh_name : string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7f3a2a51-082d-4bbc-bab2-0656f6cc3f5d",
   "metadata": {},
   "source": [
    "### 3. explode the info elements"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "843f4733-34ea-47c3-83eb-6c83977224a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import explode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "e3ee4b92-2e46-41af-9beb-a98f31db282b",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_new = df.select(\"_id\", \"district_id\", \"district_name\", \"neigh_name \", explode(\"info\").alias(\"info\")) \\\n",
    "    .select(\"_id\", \"district_id\", \"district_name\", \"neigh_name \", \"info.*\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "9bb8fc89-4572-4f6e-8d0f-0ba27ff85930",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+-------------+--------------+------+--------+----------+------------+----------+------------+----+\n",
      "|_id|district_id|district_name|   neigh_name |Amount|PerMeter|diffAmount|diffPerMeter|usedAmount|usedPerMeter|year|\n",
      "+---+-----------+-------------+--------------+------+--------+----------+------------+----------+------------+----+\n",
      "|  1|          1| Ciutat Vella|      el Raval|  97.0|  1726.5|      NULL|        NULL|      97.0|      1726.5|2013|\n",
      "|  1|          1| Ciutat Vella|      el Raval| 141.7|  2087.6|      99.1|      2534.3|     143.1|      2073.5|2014|\n",
      "|  1|          1| Ciutat Vella|      el Raval| 193.8|  2401.9|      NULL|        NULL|     193.8|      2401.9|2015|\n",
      "|  1|          1| Ciutat Vella|      el Raval| 181.0|  2805.2|      NULL|        NULL|     180.7|      2798.6|2016|\n",
      "|  1|          1| Ciutat Vella|      el Raval| 240.3|  3469.9|     292.5|      3633.1|     240.0|      3468.9|2017|\n",
      "|  2|          1| Ciutat Vella|el Barri Gòtic| 212.4|  2448.2|     285.2|      3248.7|     193.4|      2239.7|2013|\n",
      "|  2|          1| Ciutat Vella|el Barri Gòtic| 277.7|  2784.9|     240.3|      3063.2|     285.0|      2730.3|2014|\n",
      "|  2|          1| Ciutat Vella|el Barri Gòtic| 386.3|  3193.4|     355.7|      2331.6|     393.8|      3404.2|2015|\n",
      "|  2|          1| Ciutat Vella|el Barri Gòtic| 422.8|  4149.4|     409.2|      4700.7|     430.0|      3862.5|2016|\n",
      "|  2|          1| Ciutat Vella|el Barri Gòtic| 472.7|  4565.8|     799.5|      5397.4|     424.0|      4441.9|2017|\n",
      "+---+-----------+-------------+--------------+------+--------+----------+------------+----------+------------+----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_new.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "9d3ff525-aeda-4e69-8473-8f34676d9df8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _id: long (nullable = true)\n",
      " |-- district_id: long (nullable = true)\n",
      " |-- district_name: string (nullable = true)\n",
      " |-- neigh_name : string (nullable = true)\n",
      " |-- Amount: double (nullable = true)\n",
      " |-- PerMeter: double (nullable = true)\n",
      " |-- diffAmount: double (nullable = true)\n",
      " |-- diffPerMeter: double (nullable = true)\n",
      " |-- usedAmount: double (nullable = true)\n",
      " |-- usedPerMeter: double (nullable = true)\n",
      " |-- year: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_new.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "67533110-492c-4d8b-9a42-3ee845e54382",
   "metadata": {},
   "source": [
    "### 4. partition datas by district_id field for efficient query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "9182e9eb-bc02-40a3-9c2b-a487705796d1",
   "metadata": {},
   "outputs": [],
   "source": [
    "repartitioned_df = df_new.repartition(\"district_id\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4b8c2ec0-7f5e-4875-9443-36329271c213",
   "metadata": {},
   "source": [
    "### 5. save to parquet file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "682a1eec-e450-45d0-a49c-295ef9a0f515",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_path = \"./Formatted Zone/price_opendata.parquet\"\n",
    "\n",
    "repartitioned_df.write.mode(\"overwrite\").parquet(output_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2adb3b40-a02b-4e78-92e7-354090f7a32e",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
